<?php

/*
 * Pipeline
 * Author: Jeff Ober <jeffober@gmail.com>
 * Homepage: 
 *
 * Pipeline is an HttpRequestPool wrapper that largely automates one common
 * case: downloading a large number of files and calling a callback with the
 * document content.
 *
 * Example usage:
 *
 * function success($body, $req)
 * {
 *     printf("%d bytes rec'd from %s\n", strlen($body), $req->getUrl());
 * }
 *
 * function failure($code, $msg, $req)
 * {
 *     printf("HTTP %d: %s\n", $code, $msg);
 * }
 *
 * $urls = array("http://www.yahoo.com", "http://www.google.com", "http://www.msn.com", "http://www.artfulcode.net");
 * $pipe = new Pipeline(8);
 * foreach($urls as $url) $pipe->enqueue($url, 'success', 'failure');
 * $pipe->run();
 *
 * Note that in the example above, the success and failure functions may safely
 * add more work to the while the pipeline is running.
 */

/*
 * Extends HttpRequest by adding callbacks.
 */
class PipelineRequest extends HttpRequest
{
    protected $request;
    protected $onSuccess;
    protected $onError;

    public function __construct($url, $method=null, $options=null, $onSuccess=null, $onError=null)
    {
        $this->onSuccess = $onSuccess;
        $this->onError = $onError;
        parent::__construct($url, $method, $options);
    }

    /*
     * This is called by Pipeline::socketPerform on completion (successful or
     * otherwise) of a request. When the status is other than 200, the onError
     * callback (if present and callable) is run with the status code, error
     * message, and request instance as arguments. Otherwise, the onSuccess
     * method is called with the message body and request instance as
     * arguments.
     */
    public function dispatch()
    {
        $statusCode = $this->getResponseCode();

        if ($statusCode != 200) {
            if (is_callable($this->onError))
                call_user_func_array($this->onError, array($statusCode, $this->getResponseInfo('error'), $this));
            else
                trigger_error("HTTP Error {$statusCode}: {$message}", E_USER_WARNING);
        }
        else {
            if (is_callable($this->onSuccess))
                call_user_func_array($this->onSuccess, array($this->getResponseBody(), $this));
        }
    }
}

/*
 * Extends HttpRequestPool with a default method and options for requests as
 * well as the ability to limit the max number of connections at any one time.
 * The 'run' method will run all queued requests using the limit, while the
 * 'send' method (the default from HttpRequestPool) will run all at the same
 * time.
 */
class Pipeline extends HttpRequestPool
{
    protected $connections;
    protected $options;
    protected $method;
    protected $queue;

    /*
     * Creates a new Pipeline. Param $connections is an integer, defaulting to
     * 4, which limits the number of connections the Pipeline will maintain at
     * any one time. $method must be HTTP_METH_GET or another HTTP_METH_*
     * constant. $options is an array of options such as would be passed to
     * HttpRequest's constructor (see
     * http://us.php.net/manual/en/http.request.options.php).
     */
    public function __construct($connections=4, $method=HTTP_METH_GET, array $options=array())
    {
        $this->options = array_merge(array(
                'timeout' => 30,
                'connecttimeout' => 15,
                'redirect' => 3,
                'unrestrictedauth' => 1
            ), $options);
        $this->method = $method;
        $this->connections = $connections;
        $this->queue = array();
    }

    /*
     * Adds a new URL to the download queue. $onSuccess and $onError must be
     * either null or callable. $method and $options are identical to the
     * constructor parameters of the same name.
     */
    public function enqueue($url, $onSuccess=null, $onError=null, $method=null, array $options=array())
    {
        $method = is_null($method) ? $this->method : $method;
        $options = array_merge($this->options, $options);
        $request = new PipelineRequest($url, $method, $options, $onSuccess, $onError);
        array_push($this->queue, $request);
    }

    /*
     * Runs the queue, limiting the maximum number of concurrent downloads.
     */
    public function run()
    {
        $running = $this->getAttachedRequests();
        while (!empty($this->queue) || !empty($running)) {
            for ($i = $this->connections - count($running); !empty($this->queue) && $i > 0; --$i)
                $this->attach( array_shift($this->queue) );

            $this->socketPerform();
            if (!$this->socketSelect())
                throw new HttpSocketException;

            $running = $this->getAttachedRequests();
        }
    }

    /*
     * Runs the queue, ignoring the maximum number of concurrent downloads.
     */
    public function send()
    {
        while (!empty($this->queue))
            $this->attach( array_shift($this->queue) );

        parent::send();
    }

    /*
     * Each call to socketPerform will call socketPerform on the parent class
     * initially and then run the dispatch method on any completed requests.
     */
    protected function socketPerform()
    {
        try {
            $result = parent::socketPerform();
        } catch (HttpSocketException $e) {
            continue;
        }

        foreach ($this->getFinishedRequests() as $request) {
            $this->detach($request);
            $request->dispatch();
        }

        return $result;
    }
}

?>